package com.walker.util;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.jolbox.bonecp.BoneCPDataSource;

public class JdbcUtil {
	
	private static final Log log = LogFactory.getLog(JdbcUtil.class);
	
	private static BoneCPDataSource oracleBoneCP = null;
	
	/**
	 **从JDBC中获得数据库连接 
	 * @param url
	 * @param database
	 * @param username
	 * @param password
	 * @param type
	 * @return
	 */
	public static Connection getConnection(String url,String username,String password,String type) {
		Connection conn = null;
		try {
			if(StringUtils.equals(type, "mysql")) {
				//url = "jdbc:mysql://"+url+"/"+database+"?useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true";
				url = url + "?useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true";
				Class.forName("com.mysql.jdbc.Driver");
				conn = DriverManager.getConnection(url, username, password);
			}else if(StringUtils.equals(type, "oracle")){
				//url = "jdbc:oracle:thin:@"+url+":"+database;
				oracleBoneCP = new BoneCPDataSource();
				oracleBoneCP.setDriverClass("oracle.jdbc.driver.OracleDriver");
				oracleBoneCP.setJdbcUrl(url);
				oracleBoneCP.setUser(username);
				oracleBoneCP.setPassword(password);
				oracleBoneCP.setIdleConnectionTestPeriodInMinutes(60);
				oracleBoneCP.setIdleMaxAgeInMinutes(20);
				conn = oracleBoneCP.getConnection();
			}else if(StringUtils.equals(type, "hive")){
				//url = jdbc:hive2://localhost:10000
				if(!StringUtils.startsWithIgnoreCase(url,"jdbc:hive2://")){
					url = "jdbc:hive2://"+url;
				} 
				Class.forName("org.apache.hive.jdbc.HiveDriver");
				conn = DriverManager.getConnection(url,username,password);
			}else if(StringUtils.equals(type, "presto")){
				//url = jdbc:presto://localhost:9001/hive/isec_develop
				Class.forName("io.prestosql.jdbc.PrestoDriver");
				conn = DriverManager.getConnection(url,username,password);
			}else if(StringUtils.equals(type, "ck")){
				Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
				conn = DriverManager.getConnection(url,username,password);
			}  
		} catch (Exception e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(), e);
			}
			throw new RuntimeException(e.getMessage(), e);
		} 
		return conn;
	}
	
	/**
	 **获取数据库中所有的表 
	 * @param conn
	 * @param database
	 * @return
	 * @throws Exception 
	 */
	public static List<Map<String,Object>> getMyqlAllTables(Connection conn,String database) throws Exception{
		List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
		ResultSet rs = null;
		try {
			DatabaseMetaData dbMetaData = conn.getMetaData();
			rs = dbMetaData.getTables(null, database, null, new String[] {"TABLE"});
	        while(rs.next()){
				Map<String,Object> map = new HashMap<String,Object>();
				map.put("table_name", rs.getString("TABLE_NAME"));
				list.add(map);
			}
		}catch(Exception e) {
			if(log.isErrorEnabled()) {
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally {
			close(rs);
		}
		return list;
	}
	
	/**
	 ** 获取列字段信息 
	 * @param conn
	 * @param database
	 * @param table
	 * @return
	 * @throws Exception
	 */
	public static List<Map<String,Object>> getMysqlColumnInfo(Connection conn,String database,String table) throws Exception {
		String sql = "select column_name,column_type,column_comment from information_schema.columns where table_schema = ? and table_name = ? order by ordinal_position";

		List<Map<String,Object>> list = new ArrayList<Map<String, Object>>();
		PreparedStatement ps = null;
		ResultSet rs = null;
		try {
			ps = conn.prepareStatement(sql);
			ps.setObject(1,database);
			ps.setObject(2,table);
			
			rs = ps.executeQuery();
			
			list = JdbcUtil.resultSetToList(rs);
		} catch (SQLException e) {
			if(log.isErrorEnabled()) {
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally {
			close(rs);
			close(ps);
		}
		return list;
	}
	
	/**
	 **获取数据库中所有的表 
	 * @param conn
	 * @param database
	 * @return
	 * @throws Exception 
	 */
	public static List<Map<String,Object>> getOracleAllTables(Connection conn) throws Exception{
		List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
		ResultSet rs = null;
		PreparedStatement ps = null;
		try {
			String sql = "select table_name from user_tables";
			ps = conn.prepareStatement(sql);
			rs = ps.executeQuery();
			list = JdbcUtil.resultSetToList(rs);
		}catch(Exception e) {
			if(log.isErrorEnabled()) {
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally {
			close(rs);
			close(ps);
		}
		return list;
	}
	/**
	 ** 获取列字段信息 
	 * @param conn
	 * @param database
	 * @param table
	 * @return
	 * @throws Exception
	 */
	public static List<Map<String,Object>> getOracleColumnInfo(Connection conn,String user,String table) throws Exception {
		List<Map<String,Object>> list = new ArrayList<Map<String, Object>>();
		PreparedStatement ps = null;
		ResultSet rs = null;
		try {
			String sql = getOracleTableColumnSql();
			ps = conn.prepareStatement(sql);
			ps.setObject(1,user.toUpperCase());
			ps.setObject(2,table.toUpperCase());
			rs = ps.executeQuery();
			
			list = JdbcUtil.resultSetToList(rs);
		} catch (SQLException e) {
			if(log.isErrorEnabled()) {
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally {
			close(rs);
			close(ps);
		}
		return list;
	}
	
	public static boolean execSql(Connection conn,String sql) throws Exception{
    	boolean flag = true;
    	PreparedStatement ps = null;
    	try{
    		conn.setAutoCommit(false);
    		ps = conn.prepareStatement(sql);
          	ps.executeUpdate();
            conn.commit();
          	flag = true;
          }catch(SQLException e){
         	 flag = false;
              try {
                  conn.rollback();
              } catch (SQLException e1) {
            	  e1.printStackTrace();
              }
              throw new Exception(e.getMessage());
          }finally{
        	  close(ps);
          }
    	return flag ;
    }
	
	public static void execHiveSql(Connection conn, String sql) throws Exception {
		Statement stmt = null;
		try {
			stmt = conn.createStatement();
			stmt.execute(sql);
		} catch (SQLException e) {
			if (log.isErrorEnabled()) {
				log.error(e.getMessage(), e);
			}
			throw new Exception(e.getMessage());
		} finally {
			close(stmt);
		}
	}
	
	public static void execHiveSql(Connection conn, String sql,String tez_queue_name,String mapreduce_job_queuename) throws Exception {
		Statement stmt = null;
		try {
			stmt = conn.createStatement();
			if(StringUtils.isNotBlank(tez_queue_name)){
				stmt.execute("set tez.queue.name="+tez_queue_name);
			}
			if(StringUtils.isNotBlank(mapreduce_job_queuename)){
				stmt.execute("set hive.execution.engine="+mapreduce_job_queuename);
			}
			stmt.execute(sql);
		} catch (SQLException e) {
			if (log.isErrorEnabled()) {
				log.error(e.getMessage(), e);
			}
			throw new Exception(e.getMessage());
		} finally {
			close(stmt);
		}
	}
	
	/**
	 **获取数据库中所有的表 
	 * @param conn
	 * @param database
	 * @return
	 * @throws Exception 
	 */
	public static List<Map<String,Object>> queryForList(Connection conn,String sql) throws Exception{
		List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
		ResultSet rs = null;
		PreparedStatement ps = null;
		try {
			ps = conn.prepareStatement(sql);
			rs = ps.executeQuery();
			list = JdbcUtil.resultSetToList(rs);
		}catch(Exception e) {
			if(log.isErrorEnabled()) {
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally {
			close(rs);
			close(ps);
		}
		return list;
	}
	
	public static List<Map<String,Object>> queryForList(Connection conn,String sql,Object ... params) throws Exception {
		List<Map<String, Object>> list = new ArrayList<Map<String,Object>>();
		PreparedStatement ps = null;
		ResultSet rs = null;
		try {
			ps = conn.prepareStatement(sql);
			for (int j = 0; j < params.length; j++) {
				ps.setObject(j + 1, params[j]);
			}
			rs = ps.executeQuery();
			list = resultSetToList(rs);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally{
			close(rs);
			close(ps);
		}
		return list;
	}
	
	public static List<Map<String,Object>> queryForList(Connection conn,String sql,String tez_queue_name,String mapreduce_job_queuename) throws Exception {
		List<Map<String, Object>> list = new ArrayList<Map<String,Object>>();
		PreparedStatement ps = null;
		ResultSet rs = null;
		try {
			ps = conn.prepareStatement(sql);
			if(StringUtils.isNotBlank(tez_queue_name)){
				ps.execute("set tez.queue.name="+tez_queue_name);
			}
			if(StringUtils.isNotBlank(mapreduce_job_queuename)){
				ps.execute("set hive.execution.engine="+mapreduce_job_queuename);
			}
			rs = ps.executeQuery();
			list = resultSetToList(rs);
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally{
			close(rs);
			close(ps);
		}
		return list;
	}
	
	public static List<Map<String, Object>> getHiveColumnInfo(Connection conn, String database, String table) throws Exception {
		List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
		ResultSet rs = null;
		try {
			DatabaseMetaData dbMetaData = conn.getMetaData();
			rs = dbMetaData.getColumns(null, database, table, "%");
			while (rs.next()) {
				Map<String, Object> map = new LinkedHashMap<String, Object>();
				map.put("col_name", rs.getString("COLUMN_NAME"));
				map.put("data_type", rs.getString("TYPE_NAME").toLowerCase());
				map.put("comment", rs.getString("REMARKS"));
				list.add(map);
			}
		} catch (Exception e) {
			if (log.isErrorEnabled()) {
				log.error(e.getMessage(), e);
			}
			throw new Exception(e.getMessage());
		} finally {
			close(rs);
		}
		return list;
	}
	
	public static Map<String,Object> queryForMap(Connection conn,String sql,Object ... params) throws Exception {
		Map<String,Object> map = new HashMap<String,Object>();
		List<Map<String, Object>> list = queryForList(conn,sql,params);
		if(!list.isEmpty()) {
			map = list.get(0);
		}
		return map;
	}
	
	public static List<Map<String, Object>> resultSetToList(ResultSet rs) throws SQLException{
		if (rs == null) return new ArrayList<Map<String, Object>>();
		ResultSetMetaData md = rs.getMetaData(); // 得到结果集(rs)的结构信息，比如字段数、字段名等
		int columnCount = md.getColumnCount(); // 返回此 ResultSet 对象中的列数
		List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
		Map<String, Object> rowData = new HashMap<String, Object>();
		while (rs.next()) {
			rowData = new HashMap<String, Object>(columnCount);
			for (int i = 1; i <= columnCount; i++) {
				rowData.put(md.getColumnLabel(i).toLowerCase(), rs.getObject(i));
			}
			list.add(rowData);
		}
		return list;
	}
	
	/**
	 ** 批量插入
	 * @param list
	 * @param sql : insert into tbl_device(mainDevice_id,device_type,device_id,device_ip,device_port) values (?,?,?,?,?)
	 * @param columns : mainDevice_id,device_type,device_id,device_ip,device_port
	 * @param size
	 * @throws Exception 
	 */
	public static void insertBatchTableData(Connection conn,List<Map<String, Object>> list,String sql, String columns,int size) throws Exception {
		String[] columnsSplit = columns.split(","); 
		PreparedStatement ps = null;
		try {
			ps = conn.prepareStatement(sql);
			conn.setAutoCommit(false);
			int n = 0;
			for (int i = 0; i < list.size(); i++) {
				Map<String, Object> map = list.get(i);
				for (int j = 0; j < columnsSplit.length; j++) {
					ps.setObject(j + 1, map.get(columnsSplit[j]));
				}
				ps.addBatch();
				if (++n % size == 0) {
					ps.executeBatch();
					conn.commit();
				}
			}
			ps.executeBatch();
			conn.commit();
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
			throw new Exception(e.getMessage());
		}finally{
			close(ps);
		}
	}
	
	@Override
	protected void finalize() throws Throwable {
		oracleBoneCP.close();
		super.finalize();
	}

	/**
	 * 释放数据库连接
	 * 
	 * @param conn
	 */
	public static void close(Connection conn) {
		if (conn == null) {
			return;
		}
		try {
			if (conn.isClosed()) {
				return;
			}
			conn.close();
		} catch (SQLException e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(), e);
			}
		}
	}

	/**
	 * 关闭STMT
	 * @param stmt
	 */
	public static void close(Statement stmt) {
		if(stmt == null){
			return;
		}
		try {
			stmt.close();
		} catch (SQLException e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(), e);
			}
		}
	}

	/**
	 * 关闭RS
	 * 
	 * @param rs
	 */
	public static void close(ResultSet rs) {
		if(rs == null){
			return;
		}
		try {
			rs.close();
		} catch (SQLException e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(), e);
			}
		}
	}

	/**
	 * 关闭STMT和CONN
	 * @param stmt
	 * @param conn
	 */
	public static void close(Statement stmt, Connection conn) {
		close(stmt);
		close(conn);
	}

	/**
	 * 关闭RS、STMT和CONN
	 * 
	 * @param rs
	 * @param stmt
	 * @param conn
	 */
	public static void close(ResultSet rs, Statement stmt, Connection conn) {
		close(rs);
		close(stmt);
		close(conn);
	}
	
	/**
	 * 判断当前操作系统是不是window
	 * 
	 * @return boolean
	 */
	public static boolean isWindows() {
		boolean flag = false;
		if (System.getProperties().getProperty("os.name").toUpperCase().indexOf("WINDOWS") != -1) {
			flag = true;
		}
		return flag;
	}
	
	private static String getOracleTableColumnSql() {
		StringBuffer sql = new StringBuffer();
		sql.append("SELECT UCC.COLUMN_NAME AS COLUMN_NAME,");
		sql.append("      CASE WHEN UCC.DATA_TYPE = 'CHAR'     THEN UCC.DATA_TYPE||'('||UCC.CHAR_LENGTH||')'");
		sql.append("           WHEN UCC.DATA_TYPE = 'VARCHAR'  THEN UCC.DATA_TYPE||'('||UCC.CHAR_LENGTH||')'");
		sql.append("           WHEN UCC.DATA_TYPE = 'VARCHAR2' THEN UCC.DATA_TYPE||'('||UCC.CHAR_LENGTH||')'");
		sql.append("           WHEN UCC.DATA_TYPE = 'NCHAR'    THEN UCC.DATA_TYPE||'('||UCC.CHAR_LENGTH||')'");
		sql.append("           WHEN UCC.DATA_TYPE = 'NUMBER' THEN");
		sql.append("                CASE WHEN UCC.DATA_PRECISION IS NULL AND UCC.DATA_SCALE IS NULL THEN 'NUMBER'");
		sql.append("                     WHEN UCC.DATA_PRECISION IS NULL AND UCC.DATA_SCALE IS NOT NULL THEN 'NUMBER(38,'||UCC.DATA_SCALE||')'"); 
		sql.append("                     WHEN UCC.DATA_SCALE = 0 THEN  UCC.DATA_TYPE||'('||UCC.DATA_PRECISION||')'");
		sql.append("                     ELSE UCC.DATA_TYPE||'('||UCC.DATA_PRECISION||','||UCC.DATA_SCALE||')'");
		sql.append("                END");
		sql.append("           WHEN UCC.DATA_TYPE = 'NVARCHAR' THEN UCC.DATA_TYPE||'('||UCC.CHAR_LENGTH||')'");
		sql.append("           WHEN UCC.DATA_TYPE = 'NVARCHAR2' THEN UCC.DATA_TYPE||'('||UCC.CHAR_LENGTH||')'");
		sql.append("           ELSE UCC.DATA_TYPE");
		sql.append("        END COLUMN_TYPE,");
		sql.append("      ACC.COMMENTS AS COLUMN_COMMENTS");
		sql.append("   FROM ALL_TAB_COLUMNS UCC, ALL_TABLES UT, ALL_TAB_COMMENTS ATC, ALL_COL_COMMENTS ACC");
		sql.append("  WHERE UCC.OWNER = ?");
		sql.append("    AND UCC.OWNER       = UT.OWNER");
		sql.append("    AND UCC.TABLE_NAME  = UT.TABLE_NAME");
		sql.append("    AND UT.TABLESPACE_NAME IS NOT NULL");
		sql.append("    AND ATC.OWNER       = UT.OWNER");
		sql.append("    AND ATC.TABLE_NAME  = UT.TABLE_NAME");
		sql.append("    AND ACC.OWNER       = UCC.OWNER");
		sql.append("    AND ACC.TABLE_NAME  = UCC.TABLE_NAME");
		sql.append("    AND ACC.COLUMN_NAME = UCC.COLUMN_NAME");
		sql.append("    AND UCC.TABLE_NAME = ?");
		sql.append("  ORDER BY UCC.OWNER , UCC.TABLE_NAME , UCC.COLUMN_ID, UCC.COLUMN_NAME");
		return sql.toString();		
	}
	
	public static Map<String,String> parseJdbc(String jdbcUrl,String dbType) {
		Map<String,String> map = new HashMap<String, String>();
		dbType = dbType.toLowerCase();
		String prefix = null;
		String pattern = null;
		String mgrdb = null;
		switch (dbType) {
		case "oracle":
			prefix = "jdbc:oracle:thin:@";
			pattern = prefix + "(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+):(?<database>[a-zA-Z0-9_]+)?";
			mgrdb = "ORCL";
			break;
			
		case "mysql":
		case "tidb":
			prefix = "jdbc:mysql://";
			pattern = "jdbc:(?<type>[a-z]+)://(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+)/(?<database>[a-zA-Z0-9_]+)?";
			mgrdb = "information_schema";
			break;
			
		case "clickhouse":
			prefix = "jdbc:clickhouse://";
			pattern = "jdbc:(?<type>[a-z]+)://(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+)/(?<database>[a-zA-Z0-9_]+)?";
			mgrdb = "default";
			break;
			
		case "hive":
			prefix = "jdbc:hive2://";
			pattern = "jdbc:(?<type>[a-zA-Z0-9]+)://(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+)/(?<database>[a-zA-Z0-9_]+)?";
			mgrdb = "default";
			break;
			
		case "presto":
			prefix = "jdbc:presto://";
			pattern = "jdbc:(?<type>[a-z]+)://(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+)/(?<database>[a-zA-Z0-9_]+)?";
			break;
		
		case "elasticsearch":
			prefix = "http://";
			pattern = "http://(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+)";
			mgrdb = "";
			break;
			
		case "neo4j":
			prefix = "jdbc:neo4j:http://";
			pattern = "jdbc:(?<type>[a-zA-Z0-9-:])://(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+)/(?<database>[a-zA-Z0-9_]+)?";
			break;

		default:
			pattern = "jdbc:(?<type>[a-z]+)://(?<host>[a-zA-Z0-9-//.]+):(?<port>[0-9]+)/(?<database>[a-zA-Z0-9_]+)?";
			break;
		}
		
		jdbcUrl = jdbcUrl.substring(jdbcUrl.length()-4).matches("^[0-9_/]*")? jdbcUrl + "/" + mgrdb:jdbcUrl;
		jdbcUrl = jdbcUrl.startsWith(prefix)? jdbcUrl:prefix + jdbcUrl;
		jdbcUrl = jdbcUrl.endsWith("/")? jdbcUrl.substring(0,jdbcUrl.length()-1):jdbcUrl;
        Pattern namePattern = Pattern.compile(pattern);
        Matcher dateMatcher = namePattern.matcher(jdbcUrl);
        
        while (dateMatcher.find()) {
        	String host = dateMatcher.group("host");
        	String port = dateMatcher.group("port");
        	map.put("host", host);
        	map.put("port", port);
        	try {
        		map.put("database", dateMatcher.group("database"));
			} catch (Exception e) {
			}
        	map.put("url", prefix + host + ":" + port);
        	map.put("unique", host + ":" + port);
        }
        
        return map;
    }
	
	public static void main(String[] args) {
		System.out.println(parseJdbc("172.30.132.150:8123/resource", "clickhouse").get("unique"));
	}
}